package com.lagou.boot;

import com.lagou.RpcProvideApplication;
import com.lagou.client.CuratorClient;
import com.lagou.handler.UserServiceHandler;
import com.lagou.rpc.RpcDecoder;
import com.lagou.rpc.RpcEncoder;
import com.lagou.service.JSONSerializer;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.stereotype.Component;

/**
 * @author 陈浩天
 * @date 2021/2/4
 */
@Component
public class RpcProvide {
    public static void main(String[] args) throws Exception {
//        startServer("127.0.0.1",8999);
        connectZk();
    }

    //创建一个方法启动服务器
    public static void startServer(String ip, int port) throws Exception {
        //1.创建两个线程池对象
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workGroup = new NioEventLoopGroup();

        //2.创建服务端的启动引导对象
        ServerBootstrap serverBootstrap = new ServerBootstrap();

        //3.配置启动引导对象
        serverBootstrap.group(bossGroup, workGroup)
                //设置通道为NIO
                .channel(NioServerSocketChannel.class)
                //创建监听channel
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        //获取管道对象
                        ChannelPipeline pipeline = nioSocketChannel.pipeline();
                        //给管道对象pipeLine 设置编码
                        pipeline.addLast(new RpcEncoder(Object.class, new JSONSerializer()));
                        pipeline.addLast(new RpcDecoder(Object.class, new JSONSerializer()));
                        //把我们自顶一个ChannelHander添加到通道中
                        pipeline.addLast(new UserServiceHandler());
                    }
                });

        //4.绑定端口
        serverBootstrap.bind(port).sync();

        connectZk();
    }

    //注册到zookeeper上 创建 服务器节点信息
    public static void  connectZk() throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("192.168.228.130:2181") //server地址
                .sessionTimeoutMs(50000) // 会话超时时间
                .connectionTimeoutMs(30000) // 连接超时时间
                .retryPolicy(retryPolicy) // 重试策略
                .namespace("base") // ⽴命名空间/base
                .build(); //
        client.start();

        String s = client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(CuratorClient.path + "/" + RpcProvideApplication.port, String.valueOf(System.currentTimeMillis()).getBytes());

        System.out.println("节点递归创建成功， 节点路径： " + s);

        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true){
                    try {
                        Stat stat = new Stat();
                        client.getData().storingStatIn(stat).forPath(CuratorClient.path + "/" + RpcProvideApplication.port);
                        client.setData().withVersion(stat.getVersion()).forPath(CuratorClient.path + "/" + RpcProvideApplication.port, String.valueOf(System.currentTimeMillis()).getBytes());
                        Thread.sleep(5000);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        thread.start();
    }
}
